package windowtest

import Source.SourceTest.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.windowing.assigners.{EventTimeSessionWindows, SlidingEventTimeWindows, TumblingEventTimeWindows}
import org.apache.flink.streaming.api.windowing.time.Time

object WindowTest {

  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    val inputPath = "E:\\Flinklearn\\Flink\\resources\\data1\\out.txt"
    val inputStream: DataStream[String] = env.readTextFile(inputPath)

    // 转换成样例类类型
    val dataStream: DataStream[SensorReading] = inputStream.map(
      data => {
        val fields = data.split(",")
        SensorReading(fields(0), fields(1).toLong, fields(2).toDouble)
      })

    dataStream.map(data => (data.id, data.temperature))
      .keyBy(_._1)
    // 1.滚动时间窗口
    // .window(TumblingEventTimeWindows.of(Time.seconds(15),Time.seconds(2)))
    // .timeWindow(Time.seconds(15))

    // 2.滑动时间窗口
    //     .window(SlidingEventTimeWindows.of(Time.seconds(15),Time.seconds(2)))
    // .timeWindow(Time.seconds(15),Time.seconds(7))

    // 3.会话窗口
    // .window(EventTimeSessionWindows.withGap(Time.seconds(10)))

    // 4.滚动计数窗口
//      .countWindow(10)
    // 5.滑动计数窗口
      .countWindow(10,2)

  }

}
